package com.horizon.im.imServer.distributed;

import java.util.Date;
import java.util.concurrent.TimeUnit;

import com.horizon.entity.ImNode;
import com.horizon.im.common.bean.UserDTO;
import com.horizon.im.common.codec.ProtoBufDecoder;
import com.horizon.im.common.codec.ProtobufEncoder;
import com.horizon.im.imServer.serverHandler.ImNodeExceptionHandler;
import com.horizon.im.imServer.serverHandler.ImNodeHeartBeatClientHandler;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Data
public class PeerSender {
    private int reConnnectCount = 0;
    private Channel channel;

    private ImNode remoteImNode;

    private boolean connectFlag = false;
    private UserDTO user;

    private GenericFutureListener<ChannelFuture> closeListener = (ChannelFuture f) -> {
        log.info("分布式链接已经断开...", remoteImNode.toString());
        channel = null;
        connectFlag = false;
    };

    private GenericFutureListener<ChannelFuture> connectListener = (ChannelFuture f) -> {
        final EventLoop eventLoop = f.channel().eventLoop();
        if (!f.isSuccess() && ++reConnnectCount < 3) {
            log.info("链接失败，将在10s后尝试第{}次重连", reConnnectCount);
            eventLoop.schedule(() -> PeerSender.this.doConnect(), 0, TimeUnit.SECONDS);

            connectFlag = false;
        } else {
            connectFlag = true;
            log.info(new Date() + "分布式节点链接成功：{}", remoteImNode.toString());
            channel = f.channel();
            channel.closeFuture().addListener(closeListener);

            // 发送链接成功的消息

        }
    };
    private Bootstrap bootstrap;
    private EventLoopGroup eventLoopGroup;

    public PeerSender(ImNode imNode) {
        this.remoteImNode = imNode;

        bootstrap = new Bootstrap();
        eventLoopGroup = new NioEventLoopGroup();
    }

    public void doConnect() {
        String host = remoteImNode.getHost();
        int port = remoteImNode.getPort();
        if (bootstrap != null && bootstrap.group() == null) {
            bootstrap.group(eventLoopGroup);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

            bootstrap.remoteAddress(host, port);

            // 通道初始化
            bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                    nioSocketChannel.pipeline().addLast("decoder", new ProtoBufDecoder());
                    nioSocketChannel.pipeline().addLast("encoder", new ProtobufEncoder());
                    nioSocketChannel.pipeline().addLast("imNodeHeartBeatClientHandler",
                        new ImNodeHeartBeatClientHandler());
                    nioSocketChannel.pipeline().addLast("exceptionHandler", new ImNodeExceptionHandler());
                }
            });

            log.info(new Date() + "开始连接分布式节点:{}", remoteImNode.toString());
            ChannelFuture f = bootstrap.connect();
            f.addListener(connectListener);
        } else if (bootstrap.group() != null) {
            log.info(new Date() + "再一次开始连接分布式节点", remoteImNode.toString());
            ChannelFuture f = bootstrap.connect();
            f.addListener(connectListener);
        }
    }

    public void stopConnecting() {
        eventLoopGroup.shutdownGracefully();
        connectFlag = false;
    }

    public void writeAndFlush(Object pkg) {
        if (connectFlag == false) {
            log.error("分布式节点未连接:", remoteImNode.toString());
            return;
        }
        channel.writeAndFlush(pkg);
    }

}
